//package com.bsj.sipgateway.mq.consumer;
//
//import cn.hutool.cache.CacheUtil;
//import cn.hutool.cache.impl.TimedCache;
//import cn.hutool.core.util.StrUtil;
//import com.alibaba.fastjson.JSON;
//import com.bsj.sipgateway.cache.VehicleCache;
//import com.bsj.sipgateway.common.def.DvehicleInfo;
//import com.bsj.sipgateway.sip.bean.VehicleInfo;
//import com.bsj.util.PropertiesUtil;
//import lombok.extern.slf4j.Slf4j;
//import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
//import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
//import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
//import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
//import org.apache.rocketmq.client.exception.MQClientException;
//import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
//import org.apache.rocketmq.common.message.MessageExt;
//import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
//import org.springframework.stereotype.Component;
//
//import java.util.List;
//
///**
// * @Author WPH
// * @Description 车辆信息消费
// * @Date 2022/8/8 19:34
// */
//@Slf4j
//@Component
//public class VehicleConsumer {
//
//    private DefaultMQPushConsumer consumer;
//
//    private TimedCache<String, DvehicleInfo> tempCache = CacheUtil.newTimedCache(2000);
//
//    public void start() throws MQClientException {
//        subscription();
//    }
//
//    public void stop() {
//        if (consumer != null) {
//            consumer.shutdown();
//        }
//    }
//
//    public void subscription() throws MQClientException {
//        tempCache.schedulePrune(100);
//        //mq地址
//        String rocketmqAddr = PropertiesUtil.getConfig("rocketMQAddr");
//        //消费组
//        String groupName = PropertiesUtil.getConfig("rocketmqVehicleBaseInfoGroup");
//        String topic = PropertiesUtil.getConfig("rocketmqVehicleBaseInfoTopic");
//        String tag = "*";
//
//        consumer = new DefaultMQPushConsumer(groupName);
//        consumer.setNamesrvAddr(rocketmqAddr);
//
//        // 从最后开始消费
//        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
//
//        // 广播消费方式
//        consumer.setMessageModel(MessageModel.BROADCASTING);
//        consumer.subscribe(topic, tag);
//        consumer.registerMessageListener(new MessageListenerConcurrently() {
//            @Override
//            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
//                                                            ConsumeConcurrentlyContext context) {
//                for (MessageExt message : msgs) {
//                    if (null == message) {
//                        continue;
//                    }
//                    String result = new String(message.getBody());
//                    String tag = message.getTags();
//                    DvehicleInfo vehicle = null;
//                    try {
//                        switch (tag) {
//                            case "put": {
//                                vehicle = JSON.parseObject(result, DvehicleInfo.class);
//                                if (vehicle != null && ("GB28181".equals(vehicle.getTerminalType()) || "GB28181-SDK".equals(vehicle.getTerminalType()))) {
//                                    log.info("订阅到车辆基础资料信息：" + JSON.toJSONString(vehicle) + " / TAG：" + tag);
//                                    String modifyId = vehicle.getModifyId();
//                                    if (StrUtil.isNotBlank(modifyId)) {
//                                        tempCache.put(modifyId, vehicle);
//                                    }
//                                    VehicleCache.add(new VehicleInfo(vehicle));
//                                } else if (vehicle != null && !"GB28181".equals(vehicle.getTerminalType()) && !"GB28181-SDK".equals(vehicle.getTerminalType())) {
//                                    if (StrUtil.isNotBlank(vehicle.getTerminalNo()) && VehicleCache.contains(vehicle.getTerminalNo())) {
//                                        log.info("订阅到车辆基础资料信息,原来为GB28181设备修改为其他设备，删除缓存：" + JSON.toJSONString(vehicle) + " / TAG：" + tag);
//                                        VehicleCache.remove(vehicle.getTerminalNo());
//                                    }
//                                }
//                            }
//                            break;
//                            case "remove": {
//                                vehicle = JSON.parseObject(result, DvehicleInfo.class);
//                                if (vehicle != null && ("GB28181".equals(vehicle.getTerminalType()) || "GB28181-SDK".equals(vehicle.getTerminalType()))) {
//                                    if (StrUtil.isBlank(vehicle.getModifyId()) || !tempCache.containsKey(vehicle.getModifyId())) {
//                                        log.info("订阅到删除车辆基础资料信息：" + JSON.toJSONString(vehicle) + " / TAG：" + tag);
//                                        VehicleCache.remove(vehicle.getTerminalNo());
//                                    }
//                                }
//                            }
//                            break;
//                            default:
//                                //其他tags暂不处理
//                                break;
//                        }
//                    } catch (Exception e) {
//                        log.error("mq更新车辆缓存异常", e);
//                        //稍后再试
//                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
//                    }
//                }
//                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
//            }
//        });
//        consumer.start();
//        log.info("开启mq消费车辆信息成功");
//    }
//
//
//}
